package cn.doitedu.datamock;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.Executors;

public class TrafficEvents {
    public static void main(String[] args) throws InterruptedException {


        for(int i = 0;i<5;i++){
            new Thread(new Runner()).start();
            Thread.sleep(1000);
        }

    }



    public static class Runner implements Runnable{

        KafkaProducer<String, String> producer;
        JSONObject data = new JSONObject();
        JSONObject props = new JSONObject();
        //String topicName = "ods-user-action-log";
        String topicName = "backpressure";
        String[] usernames;
        String[] pages;

        public Runner(){
            Properties prop = new Properties();
            //准备服务器参数
            prop.put("bootstrap.servers","doitedu:9092");
            prop.put("acks","0");
            prop.put("retries","0");
            prop.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
            prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");


            producer = new KafkaProducer<>(prop);


            usernames = new String[]{"lvdongbing"
                    ,"windy"
                    ,"zhengsan"
                    ,"lisi"
                    ,"wangwu"
                    ,"tigertiger"
                    ,"shari"
                    ,"aewen"
                    ,"deepsea"
                    ,"hitao"
                    ,"liuyifei"
                    ,"liuyan"
                    ,"tiger"
                    ,"bird"
                    ,"uuuu"
                    ,"ooooo"
                    ,"coffe"
                    ,"retry_test"};

            pages = new String[]{
                    "/content/article/",
                    "/mall/",
                    "/mall/promotion/",
                    "/mall/search/",
                    "/waimai/shop/"
            };

        }


        @Override
        public void run() {

            for (int i = 0; i < 100000; i++) {
                data.put("user_name", usernames[RandomUtils.nextInt(0, usernames.length)]);
                data.put("lat",39.95046113360532 );
                data.put("lng",116.41900142040949);
                data.put("event_id", RandomStringUtils.randomAlphabetic(1));
                data.put("session_id", RandomStringUtils.randomAlphabetic(8));
                data.put("device_type", RandomStringUtils.randomAlphabetic(2));
                data.put("release_channel", RandomStringUtils.randomAlphabetic(2));
                data.put("action_time", System.currentTimeMillis());
                props.put("url",pages[RandomUtils.nextInt(0,pages.length)] + RandomStringUtils.randomAlphabetic(2));
                data.put("properties",props);


                ProducerRecord<String, String> record = new ProducerRecord<>(topicName, data.toJSONString());
                producer.send(record);

                if(i % 100 == 0){
                    producer.flush();
                }

                try {
                    Thread.sleep(5);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

            producer.flush();
            producer.close();


        }
    }
}
